package PassengerVehicle

import java.io.{BufferedReader, FileReader}

import org.apache.spark.sql.SparkSession

import scala.collection.immutable.HashMap

/**
  * 计算运力配置
  *
  * 四个纬度
  *   市内
  *   跨市
  *   省内
  *   省外
  *
  *   四月份
  *     incity=486609
  *     outcity=110339
  *     outprovince=10455
  *     607403
  */
object ylpz {



  def main(args: Array[String]): Unit = {
    val path = "/Users/chuxiaowei/IdeaProjects/两客一货/dataprocess/src/main/resources/base.csv"
    val cdict = readFile("/Users/chuxiaowei/IdeaProjects/两客一货/dataprocess/src/main/resources/incity.txt")
    val pdict = readFile("/Users/chuxiaowei/IdeaProjects/两客一货/dataprocess/src/main/resources/inprovince.txt")

    val spark = SparkSession.builder().master("local[4]").appName("ss").getOrCreate()
    val frame = spark.read.format("org.apache.spark.sql.execution.datasources.csv.CSVFileFormat").option("header", true).option("delimiter", ",").load(path)
    frame.registerTempTable("p")

    val map:Map[String,Int] = new HashMap[String,Int]
    val filteredframe = spark.sql("select XL,DDZ,FCRQSJ,SCKCCPHM from p where SCKCCPHM like '鲁F%' or SCKCCPHM like '鲁Y%' or SCKCCPHM is null")
    val incityValue = filteredframe.rdd.filter(row => {
      val xl = row.getString(1)
      val ddz = row.getString(2)
      matchDDZ(xl, ddz, cdict)
    }).count()

    val outcityValue = filteredframe.rdd.filter(row => {
      val xl = row.getString(1)
      val ddz = row.getString(2)
      !matchDDZ(xl, ddz, cdict) && matchDDZ(xl, ddz, pdict)
    }).count()


    val outprovinceValue = filteredframe.rdd.filter(row => {
      val xl = row.getString(1)
      val ddz = row.getString(2)
      !matchDDZ(xl, ddz, pdict)
    }).count()


    println("incity=" + incityValue)
    println("outcity=" + outcityValue)
    println("outprovince=" + outprovinceValue)

    println(incityValue + outcityValue + outprovinceValue)


  }



  def readFile(path: String):String = {
    val reader = new FileReader(path)
    val bufferReader = new BufferedReader(reader)
    val pdict = bufferReader.readLine()
    bufferReader.close()
    reader.close()
    pdict
  }


  def matchDDZ(xl: String, ddz: String, dict: String): Boolean = {
    val reg1 = "[0-9]"
    val reg2 = "（[\\u4e00-\\u9fa5]*）"
    val reg3 = "[(][0-9\\u4e00-\\u9fa5]*[)]"
    val reg4 = "服务站|下道车|机场|国际机场|车站|汽车站|长途车站|长途汽车站|中心站|中心|高速学生|高速|补|路口|总站|南站|北站|东站|西站|汽车"

    val xllen = xl.split("-").length
    if(dict.indexOf(ddz.replaceAll(reg1,"").replaceAll(reg2,"").replaceAll(reg3,"").replaceAll(reg4,"").trim) > -1) {
      true
    }else if(xllen == 2 && dict.indexOf(xl.split("-")(1).replaceAll(reg1,"").replaceAll(reg2,"").replaceAll(reg3,"").replaceAll(reg4,"").trim) > -1){
      true
    }else if (xllen == 1 && dict.indexOf(xl.replaceAll(reg1,"").replaceAll(reg2,"").replaceAll(reg3,"").replaceAll(reg4,"").trim) > -1) {
      true
    }else if (xllen == 3 && dict.indexOf(xl.split("-")(2).replaceAll(reg1,"").replaceAll(reg2,"").replaceAll(reg3,"").replaceAll(reg4,"").trim) > -1) {
      true
    }else {
      false
    }
  }

}
